package com.suxin.mqtt.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.*;

/**
 * @Classname MqttInputConfig
 * @Description [ 消费者 ]
 * @Author Tang
 * @Date 2020/10/4 20:20
 * @Created by ASUS
 */
@Configuration
public class MqttInputConfig {

    public MqttPahoMessageDrivenChannelAdapter adapter;

    @Bean
    public MessageProducer inbound() {
        adapter = new MqttPahoMessageDrivenChannelAdapter("tcp://192.168.108.128:1883", "testClient",
                         "mytopic");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);

        // 设置订阅通道
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * inputChannel = 订阅的bean的名字
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                MessageHeaders headers = message.getHeaders();
                headers.forEach((key, value) -> {
                    System.out.println(key + "--->" + value);
                });
                String topic = headers
                        .get("mqtt_receivedTopic").toString();
                System.out.println(topic + " 接收消息为---》" + message.getPayload());
            }

        };
    }

    public MqttPahoMessageDrivenChannelAdapter getAdapter() {
        return adapter;
    }

}